package wangZhong.app.dwd.db;
/**
 * @author wangz
 * @date 2022/9/2 17:09
 */

import wangZhong.app.BaseSqlApp;
import wangZhong.common.Constant;
import wangZhong.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

class Dwd_TradeOrderDetail extends BaseSqlApp {
    public static void main(String[] args) throws Exception {
        new Dwd_TradeOrderDetail().init(
                3006,
                2,
                "Dwd_TradeOrderDetail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //读取预处理表
        tEnv.executeSql("" +
                "create table dwd_trade_order_pre_process(\n" +
                "    id string,\n" +
                "    order_id string,\n" +
                "    user_id string,\n" +
                "    order_status string,\n" +
                "    course_id string,\n" +
                "    course_name string,\n" +
                "    province_id bigint,\n" +
                "    province_name string,\n" +
                "    origin_amount string,\n" +
                "    coupon_reduce string,\n" +
                "    final_amount string,\n" +
                "    date_id string,\n" +
                "    create_time string,\n" +
                "    update_time_id string,\n" +
                "    update_time string,\n" +
                "    `type` string,\n" +
                "    `old` map<string, string>,\n" +
                "    od_ts bigint,\n" +
                "    oi_ts bigint,\n" +
                "    row_op_ts timestamp_ltz(3)\n" +
                ")" + SQLUtil.getKafkaSource(Constant.TOPIC_DWD_TRADE_ORDER_PRE_PROCESS,"Dwd_TradeOrderDetail")
        );

        //过滤订单详情
        Table result = tEnv.sqlQuery(
                "select\n" +
                        "id,\n" +
                        "order_id, \n" +
                        "user_id, \n" +
                        "order_status, \n" +
                        "course_id, \n" +
                        "course_name, \n" +
                        "province_id, \n" +
                        "province_name, \n" +
                        "origin_amount, \n" +
                        "coupon_reduce, \n" +
                        "final_amount, \n" +
                        "date_id, \n" +
                        "create_time,\n" +
                        "oi_ts ts, \n" +
                        "row_op_ts \n" +
                        "from dwd_trade_order_pre_process\n" +
                        "where `type`= 'insert'"
        );

        //写到kafka
        tEnv.executeSql("create table dwd_trade_order_details(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "order_status string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "province_id bigint,\n" +
                "province_name string,\n" +
                "origin_amount string,\n" +
                "coupon_reduce string,\n" +
                "final_amount string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "ts bigint,\n" +
                "row_op_ts timestamp_ltz(3)\n" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_ORDER_DETAILS));

        result.executeInsert("dwd_trade_order_details");
    }
}
